package com.twoandone.tms.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.twoandone.tms.realtime.app.func.DwsAsyncUtil;
import com.twoandone.tms.realtime.beans.TransportTruckOrgan;
import com.twoandone.tms.realtime.utils.DateFormatUtil;
import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MyClickHouseSinkFunction;
import com.twoandone.tms.realtime.utils.ThreadPoolUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @packageName:com.twoandone.tms.realtime.app.dws
 * @ClassName:DWS
 * @Description:
 * @author:
 * @date 2023/3/11 13:48
 **/
public class DwsTransportTruckOrganWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        String topicName = "dwd_transport_task_successful";
        String groupId = "dwd_transport_task_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topicName, groupId);
        DataStreamSource<String> kafkaDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaDs");

        SingleOutputStreamOperator<TransportTruckOrgan> map = kafkaDs.map(new MapFunction<String, TransportTruckOrgan>() {
            @Override
            public TransportTruckOrgan map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String endOrgId = jsonObject.getString("end_org_id");
                String endOrgName = jsonObject.getString("end_org_name");
                String truckId = jsonObject.getString("truck_id");
                String durationTime = jsonObject.getString("duration_time");
                String actualDistance = jsonObject.getString("actual_distance");
                long ts = jsonObject.getLong("ts") * 1000;

                return TransportTruckOrgan.builder()
                        .orgId(endOrgId)
                        .orgName(endOrgName)
                        .truckId(truckId)
                        .sumSecond(Long.valueOf(durationTime))
                        .sumDistance(new BigDecimal(actualDistance))
                        .countShift(1L)
                        .ts(ts)
                        .build();
            }
        });
        SingleOutputStreamOperator<TransportTruckOrgan> operator =
                map.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<TransportTruckOrgan>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<TransportTruckOrgan>() {
                                    @Override
                                    public long extractTimestamp(TransportTruckOrgan truckOrgan, long l) {
                                        return truckOrgan.getTs();
                                    }
                                })
//                        .withIdleness(Duration.ofSeconds(20))
                );
//        operator.print("operator");
//        KeyedStream<JSONObject, String> keyByDs = map.keyBy(r->tup);


        //       异步io 关联机构
        SingleOutputStreamOperator<TransportTruckOrgan> organDs = AsyncDataStream.unorderedWait(
                operator,
                new DwsAsyncUtil<TransportTruckOrgan>("dim_base_organ") {
                    @Override
                    public void join(TransportTruckOrgan truckOrgan, JSONObject dim) {
//                        String orgParentId = dim.getString("org_parent_id");
//                        System.out.println(dim);
                        String orgLevel = dim.getString("ORG_LEVEL");
                        String regionId = dim.getString("REGION_ID");
                        if ("1".equals(orgLevel)) {
                            truckOrgan.setRegionId(regionId);
                        } else {
                            regionId = regionId.substring(0, 4) + "00";
                            truckOrgan.setRegionId(regionId);
                        }
                        String provinceId = regionId.substring(0, 2) + "0000";
                        truckOrgan.setProvinceId(provinceId);

//                        truckOrgan.setOrgParentId(orgParentId);
                        truckOrgan.setOrgLevel(orgLevel);

                    }

                    @Override
                    public String getKey(TransportTruckOrgan truckOrgan) {
                        return truckOrgan.getOrgId();
                    }

                }, 60, TimeUnit.SECONDS);
//        organDs.print("organDs");


//        关联城市
        SingleOutputStreamOperator<TransportTruckOrgan> regionDs = AsyncDataStream.unorderedWait(
                organDs,
                new DwsAsyncUtil<TransportTruckOrgan>("dim_base_region_info") {
                    @Override
                    public void join(TransportTruckOrgan jsonObject, JSONObject dim) {
//                        System.out.println("城市"+dim);
                        String id = dim.getString("ID");
                        String name = dim.getString("NAME");
                        jsonObject.setCityId(id);
                        jsonObject.setCityName(name);

                    }

                    @Override
                    public String getKey(TransportTruckOrgan truckOrgan) {

                        String regionId = truckOrgan.getRegionId();

                        return regionId;

                    }
                },
                60, TimeUnit.SECONDS);
//        regionDs.print("regionDs");
//关联省份
        SingleOutputStreamOperator<TransportTruckOrgan> provinceDs = AsyncDataStream.unorderedWait(
                regionDs,
                new DwsAsyncUtil<TransportTruckOrgan>("dim_base_region_info") {
                    @Override
                    public void join(TransportTruckOrgan jsonObject, JSONObject dim) {
//                        System.out.println("城市"+dim);

                        String provinceName = dim.getString("NAME");

                        jsonObject.setProvinceName(provinceName);

                    }

                    @Override
                    public String getKey(TransportTruckOrgan truckOrgan) {



                        return truckOrgan.getProvinceId();

                    }
                },
                60, TimeUnit.SECONDS);
//        "dim_truck_model"
//        关联卡车信息表
        SingleOutputStreamOperator<TransportTruckOrgan> truckDs = AsyncDataStream.unorderedWait(
                provinceDs,
                new DwsAsyncUtil<TransportTruckOrgan>("dim_truck_info") {
                    @Override
                    public void join(TransportTruckOrgan jsonObject, JSONObject dim) {
//                        System.out.println("truckDs:::"+dim);
//                        String modelName = dim.getString("TRUCK_MODEL_NAME");
                        String modelId = dim.getString("TRUCK_MODEL_ID");
                        jsonObject.setTruckModelId(modelId);
                    }

                    @Override
                    public String getKey(TransportTruckOrgan truckOrgan) {
                        return truckOrgan.getTruckId();
                    }
                }, 60, TimeUnit.SECONDS);
//        truckDs.print("truckDs");

        SingleOutputStreamOperator<TransportTruckOrgan> truckModuleDs = AsyncDataStream.unorderedWait(
                truckDs,
                new DwsAsyncUtil<TransportTruckOrgan>("dim_truck_model") {
                    @Override
                    public void join(TransportTruckOrgan jsonObject, JSONObject dim) {
//                        System.out.println("truckDs:::"+dim);
                        String modelName = dim.getString("MODEL_NAME");
                        jsonObject.setTruckModelName(modelName);
                    }

                    @Override
                    public String getKey(TransportTruckOrgan truckOrgan) {
                        return truckOrgan.getTruckModelId();
                    }
                }, 60, TimeUnit.SECONDS);


//        开窗聚合
        KeyedStream<TransportTruckOrgan, Tuple3<String, String, String>> keyedStream =
                truckModuleDs.keyBy(new KeySelector<TransportTruckOrgan, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(TransportTruckOrgan truckOrgan) throws Exception {
                        return Tuple3.of(truckOrgan.getOrgId(), truckOrgan.getCityId(), truckOrgan.getTruckModelId());
                    }
                });


        WindowedStream<TransportTruckOrgan, Tuple3<String, String, String>, TimeWindow> window =
                keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));


        SingleOutputStreamOperator<TransportTruckOrgan> reduce =
                window.reduce(
                        new ReduceFunction<TransportTruckOrgan>() {
                            @Override
                            public TransportTruckOrgan reduce(TransportTruckOrgan truckOrgan, TransportTruckOrgan t1) throws Exception {
                                truckOrgan.setCountShift(truckOrgan.getCountShift() + t1.getCountShift());
                                truckOrgan.setSumDistance(truckOrgan.getSumDistance().add(t1.getSumDistance()));
                                truckOrgan.setSumSecond(truckOrgan.getSumSecond() + t1.getSumSecond());
                                return truckOrgan;
                            }
                        },
                        new WindowFunction<TransportTruckOrgan, TransportTruckOrgan, Tuple3<String, String, String>, TimeWindow>() {
                            @Override
                            public void apply(Tuple3<String, String, String> stringStringStringTuple3, TimeWindow timeWindow, Iterable<TransportTruckOrgan> iterable, Collector<TransportTruckOrgan> collector) throws Exception {
                                for (TransportTruckOrgan transportTruckOrgan : iterable) {
                                    transportTruckOrgan.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                                    transportTruckOrgan.setEtt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                                    transportTruckOrgan.setTs(System.currentTimeMillis());
                                    collector.collect(transportTruckOrgan);
                                }
                            }
                        });

//        reduce.print(">>>>>");
//        写入clickhouse
        reduce.addSink(new MyClickHouseSinkFunction().getSinkFunction("insert into dws_transport_truck_organ_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
